package myapps3;

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 消费者拦截器
 */
public class ConsumerInterceptorTTL  implements ConsumerInterceptor<String,String>{
	private static final long EXPIRE_INTERVAL = 10 * 1000;

	@Override
	public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
		System.out.println("before:" + records);
		long now = System.currentTimeMillis();
		Map<TopicPartition,List<ConsumerRecord<String,String>>> newRecords = new HashMap<>();

		for (TopicPartition tp : records.partitions()) {
			List<ConsumerRecord<String, String>> tpRecords = records.records(tp);
			List<ConsumerRecord<String,String>> newTpRecords = new ArrayList<>();
			for (ConsumerRecord<String, String> tpRecord : tpRecords) {
				if(now - tpRecord.timestamp() < EXPIRE_INTERVAL){
					newTpRecords.add(tpRecord);
				}
			}

			if(!newTpRecords.isEmpty()){
				newRecords.put(tp,newTpRecords);
			}
		}
		return new ConsumerRecords<>(newRecords);
	}

	@Override
	public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
		offsets.forEach((tp,offset)-> System.out.println(tp+":"+offset.offset()));
	}

	@Override
	public void close() {

	}

	@Override
	public void configure(Map<String, ?> configs) {

	}
}
